package main

import (
	"context"
	"flag"
	"fmt"
	"github.com/prometheus/client_golang/prometheus/promhttp"
	"github.com/zeromicro/go-zero/core/conf"
	"github.com/zeromicro/go-zero/core/logx"
	"github.com/zeromicro/go-zero/core/service"
	"github.com/zeromicro/go-zero/zrpc"
	"google.golang.org/grpc"
	"google.golang.org/grpc/reflection"
	"net/http"
	"os"
	"time"
	"yunzhan/common/models"
	"yunzhan/rpc-server/internal/config"
	alertmanagerdeployServer "yunzhan/rpc-server/internal/server/alertmanager_deploy"
	checkcomponentServer "yunzhan/rpc-server/internal/server/check_component"
	elasticsearchServer "yunzhan/rpc-server/internal/server/elasticsearch_deploy"
	flinkdeployServer "yunzhan/rpc-server/internal/server/flink_deploy"
	grafanadeployServer "yunzhan/rpc-server/internal/server/grafana_deploy"
	hbasedeployServer "yunzhan/rpc-server/internal/server/hbase_deploy"
	hdfsdeployServer "yunzhan/rpc-server/internal/server/hdfs_deploy"
	heartServer "yunzhan/rpc-server/internal/server/heart"
	hivedeployServer "yunzhan/rpc-server/internal/server/hive_deploy"
	javaserver "yunzhan/rpc-server/internal/server/java_deploy"
	kafkaserver "yunzhan/rpc-server/internal/server/kafka_deploy"
	keepaliveddeployServer "yunzhan/rpc-server/internal/server/keepalived_deploy"
	miniodeployServer "yunzhan/rpc-server/internal/server/minio_deploy"
	nginxdeployServer "yunzhan/rpc-server/internal/server/nginx_deploy"
	prometheusdeployServer "yunzhan/rpc-server/internal/server/prometheus_deploy"
	redisdeployServer "yunzhan/rpc-server/internal/server/redis_deploy"
	sparkdeployServer "yunzhan/rpc-server/internal/server/spark_deploy"
	systembasicconfigServer "yunzhan/rpc-server/internal/server/system_basic_config"
	yarndeployServer "yunzhan/rpc-server/internal/server/yarn_deploy"
	zookeeperserver "yunzhan/rpc-server/internal/server/zookeeper_deploy"
	"yunzhan/rpc-server/internal/svc"
	agent "yunzhan/rpc-server/pb"
)

var configFile = flag.String("f", "etc/config.yaml", "the config file")

func main() {
	flag.Parse()

	var c config.Config
	conf.MustLoad(*configFile, &c)
	logx.MustSetup(c.ServicesLogConfig["agentServer"].LogConf)

	if c.ServicesLogConfig["agentServer"].LogConf.Mode == "file" {
		logFilePath := fmt.Sprintf(`%s/agent-server.log`, c.ServicesLogConfig["agentServer"].LogConf.Path)
		file, err := os.OpenFile(logFilePath, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
		if err != nil {
			logx.Error(err)
			return
		}
		defer file.Close()
		writer := logx.NewWriter(file)
		logx.SetWriter(writer)
	}

	defer func() {
		err := logx.Close()
		if err != nil {
			panic(err)
		}
	}()
	ctx := svc.NewServiceContext(c)

	heartbeatCtx, cancel := context.WithCancel(context.Background())
	defer cancel()

	// Set up Prometheus metrics handler
	http.Handle("/metrics", promhttp.Handler())
	go func() {
		logx.Infof(fmt.Sprintf("Starting Prometheus metrics server at http://localhost:%d/metrics", c.AgentConf.MetricsPort))
		if err := http.ListenAndServe(fmt.Sprintf(":%d", c.AgentConf.MetricsPort), nil); err != nil {
			logx.Errorf("Failed to start Prometheus metrics server: %v", err)
		}
	}()

	go startHeartbeatTask(heartbeatCtx, fmt.Sprintf("%s:%d", c.RestServices["apiServer"].RestConf.Host, c.RestServices["apiServer"].RestConf.Port))

	s := zrpc.MustNewServer(c.RpcService.RpcServerConf, func(grpcServer *grpc.Server) {
		agent.RegisterHeartServer(grpcServer, heartServer.NewHeartServer(ctx))
		agent.RegisterJavaDeployServer(grpcServer, javaserver.NewJavaDeployServer(ctx))
		agent.RegisterZooKeeperDeployServer(grpcServer, zookeeperserver.NewZooKeeperDeployServer(ctx))
		agent.RegisterKafkaDeployServer(grpcServer, kafkaserver.NewKafkaDeployServer(ctx))
		agent.RegisterRedisDeployServer(grpcServer, redisdeployServer.NewRedisDeployServer(ctx))
		agent.RegisterGrafanaDeployServer(grpcServer, grafanadeployServer.NewGrafanaDeployServer(ctx))
		agent.RegisterPrometheusDeployServer(grpcServer, prometheusdeployServer.NewPrometheusDeployServer(ctx))
		agent.RegisterAlertManagerDeployServer(grpcServer, alertmanagerdeployServer.NewAlertManagerDeployServer(ctx))
		agent.RegisterHdfsDeployServer(grpcServer, hdfsdeployServer.NewHdfsDeployServer(ctx))
		agent.RegisterYarnDeployServer(grpcServer, yarndeployServer.NewYarnDeployServer(ctx))
		agent.RegisterFlinkDeployServer(grpcServer, flinkdeployServer.NewFlinkDeployServer(ctx))
		agent.RegisterSparkDeployServer(grpcServer, sparkdeployServer.NewSparkDeployServer(ctx))
		agent.RegisterMinIODeployServer(grpcServer, miniodeployServer.NewMinIODeployServer(ctx))
		agent.RegisterHBaseDeployServer(grpcServer, hbasedeployServer.NewHBaseDeployServer(ctx))
		agent.RegisterHiveDeployServer(grpcServer, hivedeployServer.NewHiveDeployServer(ctx))
		agent.RegisterElasticsearchDeployServer(grpcServer, elasticsearchServer.NewElasticsearchDeployServer(ctx))
		agent.RegisterNginxDeployServer(grpcServer, nginxdeployServer.NewNginxDeployServer(ctx))
		agent.RegisterKeepalivedDeployServer(grpcServer, keepaliveddeployServer.NewKeepalivedDeployServer(ctx))
		agent.RegisterSystemBasicConfigServer(grpcServer, systembasicconfigServer.NewSystemBasicConfigServer(ctx))
		agent.RegisterCheckComponentServer(grpcServer, checkcomponentServer.NewCheckComponentServer(ctx))

		if c.RpcService.Mode == service.DevMode || c.RpcService.Mode == service.TestMode {
			reflection.Register(grpcServer)
		}
	})

	logx.Infof("Starting %s server at http://%s", c.RpcService.Name, c.RpcService.ListenOn)
	s.Start()
}

func startHeartbeatTask(ctx context.Context, url string) {
	ticker := time.NewTicker(10 * time.Second)
	defer ticker.Stop()

	for {
		select {
		case <-ticker.C:
			heartbeatInfo, err := models.GetHeartbeatInfo()
			if err != nil {
				logx.Errorf("Failed to get heartbeat info: %v", err)
				continue
			}
			err = models.SendHeartbeat(ctx, url, heartbeatInfo)
			if err != nil {
				logx.Errorf("Failed to send heartbeat info: %v", err)
			}
		case <-ctx.Done():
			logx.Infof("Heartbeat task stopped.")
			return
		}
	}
}
